package com.mjk.common.tools.scheduled.single.impl;


import com.mjk.common.core.dcc.DccClient;
import com.mjk.common.core.dcc.DccSessionHolder;
import com.mjk.common.core.dcc.DccWatcher;
import com.mjk.common.tools.logger.BizLogger;
import com.mjk.common.tools.scheduled.single.SinglePointInterface;
import com.mjk.common.util.ObjectUtils;
import com.mjk.common.util.ThreadUtils;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;

import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
 * 基于ZK的单点运行判断
 * @author mjk
 *
 */
public class ZKSingelPointImpl extends DccClient implements SinglePointInterface,ApplicationContextAware{
	
	@Value(value="${server.port:8080}")
	private String nodeName ;

    final Map<String,DataSingleListener> listeners = new ConcurrentHashMap<>();
	
	public ZKSingelPointImpl(String zkHost, int timeout) {		
		super(zkHost, timeout);
		try{
            ZooKeeper zk = getHolder().connect();
			//初始化根节点
			Stat stat = zk.exists(ROOT_PATH, false);
			if (stat == null) {
				zk.create(ROOT_PATH, new byte[0],
						ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
			}
			//节点监听
			registNodeListener(ROOT_PATH, new DccWatcher.NodeListener() {
				@Override
				public void onRefreshChildren(List<String> children) {
					//选举跑sechedule的节点
					String leader = Vote.vote(children);
					writeData(ROOT_PATH, leader);
				}			
			});
		}catch(Exception e){
			BizLogger.error(e);
			Thread.currentThread().interrupt();
		}
	}
	
	@Override
	public boolean isSingelRun() {
		String dccDate = readData(ROOT_PATH);
        return nodeName.equals(dccDate.replace("\"", ""));
    }
	/**
	 * 节点值监听
	 */
	@Override
	public void registerListener(String node,DataSingleListener listener) {
		//数据监听
		node = rightPath(node);
		listeners.put(node, listener);
		registDataListener(node, new DccWatcher.DataListener(){
			@Override
			public void onUpdate(String path, Object data) {				
				if(listeners.containsKey(path)){
					listeners.get(path).onListener(rmRightPath(path),vaildData((String)data));
				}
			}
		});
		registNodeListener(node, new DccWatcher.NodeListener() {
		});
	}
	
	@Override
	public void setNotic(String node, String value) {
		writeData(node, value,false);
			
	}
	/**
	 * 设置通知，带超时时间
	 */
	@Override
	public void setNotic(String node, String value,long timeOut) {
		if(timeOut>0){
			timeOutExecutor.execute(() -> {
                ThreadUtils.sleep(timeOut);
                writeData(node, value,false);
            });
		}else{
			setNotic(node, value);
		}		
	}
	/**
	 * 选举类
	 * @author mjk
	 *
	 */
	static class Vote{
		private Vote(){}
		public static String vote(List<String> nodes){			
			   int candIndex = -1;
			   int k = 0;
			   for ( int i = 0; i < nodes.size()  ; i++) {
			       if ( k == 0 ){
			           candIndex = i;
			           k = 1;
			       }else{
			           if(nodes.get(candIndex).equals(nodes.get(i))) {
			               k++;
			           }else {
			               k --;
			           }
			       }          
			   }
			   return nodes.get(candIndex);             				
		}
	}
	
	@Override
	public void setApplicationContext(ApplicationContext context)throws BeansException {
		//钩子 通知 jvm退出的时候 关闭改临时节点的通知		
		Runtime.getRuntime().addShutdownHook(new SecheduleHook(getHolder()));
		writeData(ROOT_PATH+"/"+nodeName, "up",false);
	}
	
	class SecheduleHook extends Thread{
		private final DccSessionHolder holder ;
		public SecheduleHook(DccSessionHolder holder){
			this.holder = holder;
		}		
		@Override
		public void run() {
			holder.del(ROOT_PATH+"/"+nodeName);
		}
	}
	private String rmRightPath(String path) {
		if (ObjectUtils.isEmpty(path)) {
			return "";
		}
		if (path.startsWith(NODE_PATH_SEPARATOR)) {
			path = path.substring(1);
		}
		return path;
	}
	
	private  String vaildData(String data){
		if (ObjectUtils.isEmpty(data)) {
			return "";
		}
		if(data.startsWith("\"")){
			data = data.substring(1);
		}
		if(data.endsWith("\"")){
			data = data.substring(0, data.length()-1);
		}
		return data;
	}
	

}
